package com.huitone.smspfm.interfacz.mo;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.interfacz.logger.ExtLogger;
import com.huitone.smspfm.interfacz.model.MoDealHash;
import com.huitone.smspfm.interfacz.model.NotifyHash;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.context.thread.AppCustomBlockingThreadPool;
import com.huitone.smspfm.interfacz.core.AppStaticConfigs;
import com.huitone.smspfm.socket.core.handler.DefaultMessageHandler;
import com.huitone.smspfm.socket.core.net.ExtSocket;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.DeliverResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report2Resp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.ReportResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;

/**   
* @Title DelRepMessageHandler.java 
* @Package com.huitone.smspfm.interfacz.delrep 
* @Description TODO(用一句话描述该文件做什么) 
* @author ybliang   
* @date 2017年4月5日 上午10:31:43 
* @version V1.0   
*/
public class DelRepMessageHandler extends DefaultMessageHandler {
	
	private static final ExtLogger logger = ExtLogger.create(DelRepMessageHandler.class);
	
	private ExecutorService executorService = new ThreadPoolExecutor(30, 
			30, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(60), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	private ExecutorService reptEs = AppCustomBlockingThreadPool.newBlockingThreadPool(10);
	private ExecutorService delvEs = AppCustomBlockingThreadPool.newBlockingThreadPool(10);
	
	private DelRepArrayBlockingQueue reptResp = new DelRepArrayBlockingQueue(1000);
	private DelRepArrayBlockingQueue delvResp = new DelRepArrayBlockingQueue(1000);
	
	private Thread reptTh = null;
	private Thread delvTh = null;
	
	private boolean destory = false;
	
	public static long reptRespCount = 0;
	public static long reptRespCount2 = 0;
	
	public DelRepMessageHandler() {
		
		saveDelvResp();
		
		saveReptResp();
		
	}
	
	@Override
	public void handleDeliverRespMessage(Head head, DeliverResp body, ExtSocket socket) {
		
		executorService.execute(new Runnable() {
			
			@Override
			public void run() {
				
				logger.drespms(new Message(head, body).toString());
				
				String serialNo = head.getSerialNo();
				
				if (!StringUtils.isEmpty(serialNo)) {
					delvResp.put(serialNo + ":" + System.currentTimeMillis());
				}
			}
		});
	}

	@Override
	public void handleReportRespMessage(Head head, ReportResp body, ExtSocket socket) {
		
		executorService.execute(new Runnable() {
			
			@Override
			public void run() {
				
				logger.rrespms(new Message(head, body).toString());
				
				String serialNo = head.getSerialNo();
				
				if (!StringUtils.isEmpty(serialNo)) {
					reptResp.put(serialNo + ":" + System.currentTimeMillis());
				}
			}
		});
	}
	
	@Override
	public void handleReport2RespMessage(Head head, Report2Resp body, ExtSocket socket) {
		
		executorService.execute(new Runnable() {
			
			@Override
			public void run() {
				
				logger.rrespms(new Message(head, body).toString());
				
				String serialNo = head.getSerialNo();
				
				if (!StringUtils.isEmpty(serialNo)) {
					reptResp.put(serialNo + ":" + System.currentTimeMillis());
				}
			}
		});
	}
	
	/**
	 * 保存上行短信推送响应信息
	 */
	public void saveDelvResp() {
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> delvScript = new DefaultRedisScript<>(AppStaticConfigs.luaMoRespSucc, List.class);
		
		ArrayList<String> delvKeys = new ArrayList<>();
		delvKeys.add(AppStaticConfigs.iSMMoRespSerialNoZSet);
		delvKeys.add(AppStaticConfigs.iSMMoDealSerialNoSet);
		//delvKeys.add(AppStaticConfigs.iSMMoEndList);
		
		delvTh = new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while((executorService != null && !executorService.isShutdown()) || !destory || delvResp.getSize() > 0) {
					
					delvEs.execute(new Runnable() {
						
						@Override
						public void run() {
							
							if ((executorService != null && executorService.isShutdown()) && destory && delvResp.getSize() <= 0) return; 
							
							List<String> list = delvResp.poll();
							
							if (list != null && !list.isEmpty()) {
								
								List<List<String>> entities = AppStaticConfigs.scriptOpts2.getHashList(delvScript, delvKeys, list.toArray());
								
								List<MoDealHash> lists = new ArrayList<MoDealHash>();
								for (int i = 0 ; i < entities.size() ; i ++) {
									List<String> tmpList = entities.get(i);
									MoDealHash entity = MoDealHash.parseStr(tmpList);
									lists.add(entity);
								}
								
								if (!lists.isEmpty()) {
									
									AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

										@Override
										public void excute(RedisConnection connection) {
											
											List<byte[]> moendList = new ArrayList<>();
											
											for (MoDealHash mdh:lists) {
												
												Map<byte[], byte[]> map = new HashMap<>();
												
												map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
												map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + mdh.getiSMCacheKey()), map);
												
												moendList.add(AppStaticConfigs.stringRedisSerializer.serialize(mdh.getiSMCacheKey()));
											}
											
											if (!moendList.isEmpty()) {
												byte[][] bs = moendList.toArray(new byte[1][]);
												connection.lPush(AppStaticConfigs.iSMMoEndListB, bs);
											}
										}
									});
								}
							} else {
								try {
									TimeUnit.MILLISECONDS.sleep(10);
								} catch (InterruptedException e) {}
							}
						}
					});
				}
				
				delvEs.shutdown();
			}
		});
		delvTh.start();
		
	}
	
	/**
	 * 保存状态报告推送响应信息
	 */
	public void saveReptResp() {
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> reptScript = new DefaultRedisScript<>(AppStaticConfigs.luaRptRespSucc, List.class);
		
		ArrayList<String> reptKeys = new ArrayList<>();
		reptKeys.add(AppStaticConfigs.iSMRptRespSerialNoZSet);
		reptKeys.add(AppStaticConfigs.iSMNotifySerialNoSet);
		
		reptTh = new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while((executorService != null && !executorService.isShutdown()) || !destory || reptResp.getSize() > 0) {
					
					reptEs.execute(new Runnable() {
						
						@Override
						public void run() {
							
							if ((executorService != null && executorService.isShutdown()) && destory && reptResp.getSize() <= 0) return;
							
							List<String> list = reptResp.poll();
							
							if (list != null && !list.isEmpty()) {
								
								List<List<String>> entities = AppStaticConfigs.scriptOpts2.getHashList(reptScript, reptKeys, list.toArray());
								
								List<NotifyHash> lists = new ArrayList<NotifyHash>();
								for (int i = 0 ; i < entities.size() ; i ++) {
									List<String> tmplist = entities.get(i);
									NotifyHash entity = NotifyHash.parseStr(tmplist);
									lists.add(entity);
								}
								
								if (!lists.isEmpty()) {
									
									AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

										@Override
										public void excute(RedisConnection connection) {
											
											List<byte[]> notifyendList = new ArrayList<>();
											
											for (NotifyHash nfh:lists) {
												
												int iNotifyType = nfh.getiNotifyType();
												
												Map<byte[], byte[]> map = new HashMap<>();
												
												if (iNotifyType == 1) {
													map.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
													map.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
												} else {
													map.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
													map.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
												}
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + nfh.getiSMCacheKey()), map);
												
												notifyendList.add(AppStaticConfigs.stringRedisSerializer.serialize(nfh.getiSMCacheKey() + ":" + iNotifyType));
											}
											
											if (!notifyendList.isEmpty()) {
												byte[][] bs = notifyendList.toArray(new byte[1][]);
												connection.lPush(AppStaticConfigs.iSMNotifyEndListB, bs);
											}
										}
									});
								}
							} else {
								try {
									TimeUnit.MILLISECONDS.sleep(10);
								} catch (InterruptedException e) {}
							}
						}
					});
				}
				
				reptEs.shutdown();
			}
		});
		reptTh.start();
	}

	@Override
	public void destroy() {
		
		if (executorService != null && !executorService.isShutdown()) {
			executorService.shutdown();
		}
		
		while (executorService != null && !executorService.isShutdown()) {
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {}
		}
		
		if (executorService != null && executorService.isShutdown()) {
			executorService = null;
		}
		
		if (reptResp != null) {
			reptResp.destroy();
		}
		
		if (delvResp != null) {
			delvResp.destroy();
		}
		
		destory = true;
	}

}
